package com.cl.ks.flow;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.chlorine.base.util.ChineseUtil;
import com.cl.ks.entity.*;
import com.cl.ks.flow.base.BaseFlowNodeHandler;
import com.cl.ks.service.ClueAutoResultService;
import com.cl.ks.service.KsFlowNodeResultService;
import com.cl.ks.utils.QueryDslUtil;
import com.cl.spark.dto.SparkResult;
import com.cl.ks.DTO.Progress;
import com.cl.ks.flow.factory.KsFlowGraphNodeFactory;
import com.cl.ks.repository.KsFlowGraphRepository;
import com.cl.ks.service.KsFlowGraphRecordService;
import com.querydsl.core.Tuple;
import com.querydsl.core.types.dsl.SimpleExpression;
import com.querydsl.jpa.impl.JPAQueryFactory;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.map.LinkedMap;
import org.apache.commons.lang3.StringUtils;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;

import static com.cl.ks.constant.QTableConstant.qKsFlowGraphLog;
import static com.cl.ks.constant.QTableConstant.qKsFlowGraphRecord;

/**
 * 流程图处理器入口类
 */
@Component
@Slf4j
public class KsFlowGraphHandler {

    private final static Map<String, LinkedHashMap<String, Map<String, Object>>> progressMap = new HashMap<>();

    private final static Set<Integer> runningSet = new HashSet<>();

    private final KsFlowGraphRecordService ksFlowGraphRecordService;

    private final KsFlowNodeResultService ksFlowNodeResultService;

    private final KsFlowGraphNodeFactory ksFlowGraphNodeFactory;
    private final KsFlowGraphRepository ksFlowGraphRepository;

    private final JPAQueryFactory queryFactory;
    private final ClueAutoResultService clueAutoResultService;


    public KsFlowGraphHandler(KsFlowGraphRecordService ksFlowGraphRecordService, KsFlowNodeResultService ksFlowNodeResultService, KsFlowGraphNodeFactory ksFlowGraphNodeFactory,
                              KsFlowGraphRepository ksFlowGraphRepository, JPAQueryFactory queryFactory, ClueAutoResultService clueAutoResultService) {
        this.ksFlowGraphRecordService = ksFlowGraphRecordService;
        this.ksFlowNodeResultService = ksFlowNodeResultService;
        this.ksFlowGraphNodeFactory = ksFlowGraphNodeFactory;
        this.ksFlowGraphRepository = ksFlowGraphRepository;
        this.queryFactory = queryFactory;
        this.clueAutoResultService = clueAutoResultService;
    }


    /**
     * @param ksFlowGraph 流程图配置
     * @return 结束节点，包含执行结果
     * 1. 传入流程图配置，存储执行记录
     * 2. 从开始节点使用广度优先搜索逐层遍历
     * 3. 根据节点名称从节点工厂获取对应的节点处理器，对输入数据进行处理
     * 4. 处理完成后输出，每个节点的输出数据用于下一个节点的输入数据
     * 5. 遍历到结束节点时，存储输出结果和日志
     */
    @Async("taskExecutor")
    public CompletableFuture<SparkNode> handleKsFlowGraph(KsFlowGraph ksFlowGraph) {
        if (!ksFlowGraph.getChildGraphFlag() && isKsFlowGraphRunning(ksFlowGraph.getId())) {
            return null;
        }
        runningSet.add(ksFlowGraph.getId());
        try {
            KsFlowNodeResult deleteParam = new KsFlowNodeResult();
            deleteParam.setKsFlowGraphId(ksFlowGraph.getId());
            ksFlowNodeResultService.deleteAll(deleteParam);
            KsFlowGraphRecord ksFlowGraphRecord = ksFlowGraphRecordService.saveKsFlowGraphRecord(ksFlowGraph);

            CompletableFuture<SparkNode> result = null;
            SparkNode startNode = null;
            if (ksFlowGraph.getSparkNode() != null) {
                startNode = ksFlowGraph.getSparkNode();
            } else {
                startNode = JSON.parseObject(ksFlowGraph.getSparkNodeJson(), SparkNode.class);
            }
            if (startNode != null ) {
                String variableJson = ksFlowGraph.getVariableJson();
                if (StringUtils.isNotBlank(variableJson) && variableJson.length() > 23) {
                    String startNodeStr = JSONObject.toJSONString(startNode);
                    List<Variable> variables = JSONArray.parseArray(variableJson, Variable.class);
                    for (Variable variable : variables) {
                        startNodeStr = startNodeStr.replace(variable.key, variable.value);
                    }
                    startNode = JSON.parseObject(startNodeStr, SparkNode.class);
                }
                SparkNode endNode = null;
                Map<String, SparkNode> sparkNodeMap = new HashMap<>();// 用于广度优先遍历
                Queue<SparkNode> queue = new LinkedList<>();// 用于判断是否遍历过不同节点的共有子节点
                Map<String, Integer> nodeAdded = new HashMap<>();
                queue.offer(startNode);
                while (!queue.isEmpty()) {
                    SparkNode currentNode = queue.poll();
                    endNode = currentNode;
                    SparkResult sparkResult = handleSparkNode(new BaseFlowNodeHandler.ProcessParam(currentNode, ksFlowGraph, ksFlowGraphRecord, new KsFlowGraphLog(), ksFlowGraph.getTestFlag()));
                    log.info(currentNode.getNodeName());
                    if (currentNode.getNodePath() != null) {
                        log.info(currentNode.getNodePath());
                    }

                    List<SparkNode> nextNodes = currentNode.getNextSparkNodes();
                    if (nextNodes == null) {
                        continue;
                    }
                    for (SparkNode nextNode : nextNodes) {
                        if (nextNode.getCode() != null && sparkNodeMap.containsKey(nextNode.getCode())) {
                            nextNode = sparkNodeMap.get(nextNode.getCode());
                        }
                        if (nextNode.getInputSparkResultList() == null) {
                            nextNode.setInputSparkResultList(new LinkedMap());
                        }

                        String key = generateInputDataKey(currentNode);

                        nextNode.getInputSparkResultList().put(key, sparkResult);
                        if (nextNode.getCode() != null) {
                            sparkNodeMap.put(nextNode.getCode(), nextNode);
                        }
                        if (!nodeAdded.containsKey(nextNode.getCode())) {
                            nodeAdded.put(nextNode.getCode(), 1);
                        } else {
                            nodeAdded.put(nextNode.getCode(), nodeAdded.get(nextNode.getCode()) + 1);
                        }
                        if (Objects.equals(nodeAdded.get(nextNode.getCode()), nextNode.getInputNodeCount())) {
                            queue.offer(nextNode);
                        }
                    }
                }
                result = CompletableFuture.completedFuture(endNode);
            }

            ksFlowGraphRecord.setEndTime(LocalDateTime.now());
            ksFlowGraphRecord.setStatus(1);
            ksFlowGraphRecordService.save(ksFlowGraphRecord);
            return result;
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            runningSet.remove(ksFlowGraph.getId());
        }

        return null;
    }

    public SparkResult handleSparkNode(BaseFlowNodeHandler.ProcessParam processParam) {
        SparkNode sparkNode = processParam.getSparkNode();
        String nodeHandlerName = sparkNode.getNodeName();
        if (sparkNode.getNodeExpression() != null && sparkNode.getNodeExpression().containsKey("nodeHandler")) {
            nodeHandlerName = sparkNode.getNodeExpression().getString("nodeHandler");
        }
        return ksFlowGraphNodeFactory.getKsFlowGraphNodeHandler(nodeHandlerName).handle(processParam);
    }

    public List<Map<String, Object>> getProgress(Integer ksFlowGraphId) {
        progressMap.putIfAbsent(String.valueOf(ksFlowGraphId), new LinkedHashMap<>());
        List<Map<String, Object>> list = getLogList(ksFlowGraphId);
        list.forEach(r -> progressMap.get(String.valueOf(ksFlowGraphId)).put((String) r.get("nodeCode"), r));
        return list;
    }

    public List<Map<String, Object>> getLogList(Integer ksFlowGraphId) {
        SimpleExpression[] fields = {qKsFlowGraphLog.createTime.as("createTime"),
                qKsFlowGraphLog.endTime.as("endTime"),
                qKsFlowGraphLog.inputSize.as("inputSize"),
                qKsFlowGraphLog.outputSize.as("outputSize"),
                qKsFlowGraphLog.msg.as("msg"),
                qKsFlowGraphLog.sparkNodeName.as("nodeName"),
                qKsFlowGraphLog.sparkNodeCode.as("nodeCode"),
                qKsFlowGraphLog.status.as("status")};
        Integer ksFlowGraphRecordId = queryFactory.select(qKsFlowGraphRecord.id.max()).from(qKsFlowGraphRecord).where(qKsFlowGraphRecord.ksFlowGraphId.eq(ksFlowGraphId)).fetchFirst();
        if (ksFlowGraphRecordId == null) {
            return new ArrayList<>();
        }
        List<Tuple> ksFlowGraphLogList = queryFactory.select(fields).from(qKsFlowGraphLog).where(qKsFlowGraphLog.ksFlowGraphId.eq(ksFlowGraphId).and(qKsFlowGraphLog.ksFlowGraphRecordId.eq(ksFlowGraphRecordId))).fetch();
        return QueryDslUtil.tupleListToMapList(ksFlowGraphLogList, fields);
    }

    public String generateInputDataKey(SparkNode currentNode) {
        String key = "";
        String nodeName = currentNode.getNodeName();
        if ("模型".equals(nodeName)) {
            key = (String) currentNode.getNodeExpression().get("ruleName");
        } else if ("数据源".equals(nodeName)) {
            key = (String) currentNode.getNodeExpression().get("targetTableName");
        } else if ("子流程".equals(nodeName)) {
            key = (String) currentNode.getNodeExpression().get("name");
        } else if (currentNode.getNodeExpression() != null) {
            key = (String) currentNode.getNodeExpression().get("name");
        }

        key += currentNode.getCode();
        StringBuilder inputDataNames = new StringBuilder();
        if (currentNode.getInputSparkResultList() != null && !currentNode.getInputSparkResultList().isEmpty()) {
            List<String> inputDataKeySet = new ArrayList<>(currentNode.getInputSparkResultList().keySet());
            for (int i = 0; i < inputDataKeySet.size(); i++) {
                inputDataNames.append("source").append(i).append(":").append(ChineseUtil.extractChinese(inputDataKeySet.get(i))).append(";");
            }
        }

        key += inputDataNames;

        return key;
    }

    public static boolean isKsFlowGraphRunning(Integer id) {
        return runningSet.contains(id);
    }
    public static boolean removeKsFlowGraphRunning(Integer id) {
        return runningSet.remove(id);
    }


    @Data
    static class Variable {
        private String key;
        private String value;
    }

}
